package air

import (
	"bytes"
	"crypto/md5"
	"encoding/base64"
	"fmt"
	"io"
	"math/rand"
	"net"
	"sync"
	"sync/atomic"
	"time"

	"gitee.com/sdxstar/air/protocol"

	"github.com/pkg/errors"
)

var (
	ErrConnClosed     = errors.New("use of closed network connection")
	ErrMessageTimeout = errors.New("timeout")
)

type Conn interface {
	Id() string
	IsClosed() bool
	Close()
	Get(key string) interface{}
	Set(key string, value interface{})
	RemoteAddr() net.Addr

	Emit(event string, v interface{}) error
	Ack(event string, v interface{}, r interface{}) error
	BroadcastToAll(event string, msg []byte)
	BroadcastTo(room, event string, msg []byte)
	Join(room string)
	Leave(room string)
}

type connection struct {
	sid                string
	srv                *Server
	conn               net.Conn
	ctx                sync.Map
	heartbeatResetChan chan struct{}

	closeFlag int32
	closeOnce sync.Once // close the conn, once, per instance
	closeChan chan struct{}

	receiveChan chan protocol.Packet
	sendChan    chan protocol.Packet

	waitReply sync.Map
}

func newConn(conn net.Conn, srv *Server) *connection {
	return &connection{
		sid:                generateSid(conn.RemoteAddr().String()),
		srv:                srv,
		conn:               conn,
		ctx:                sync.Map{},
		heartbeatResetChan: make(chan struct{}),
		closeChan:          make(chan struct{}),
		receiveChan:        make(chan protocol.Packet, srv.opts.PacketReceiveChanLimit),
		sendChan:           make(chan protocol.Packet, srv.opts.PacketSendChanLimit),
		waitReply:          sync.Map{},
	}
}

func (c *connection) Id() string {
	return c.sid
}

func (c *connection) start() {
	c.srv.waitGroup.Add(4)
	go c.checkHeartbeat()
	go c.handleLoop()
	go c.readLoop()
	go c.writeLoop()
	if h, ok := c.srv.eventHandlers[OnConnect]; ok {
		go func() {
			defer func() {
				if err := recover(); err != nil {
					c.srv.log.Error(err)
				}
			}()
			err := h(newContext(c, nil))
			if err != nil {
				c.srv.log.Error(err)
				c.Close()
			}
		}()
	}
}

func (c *connection) Close() {
	c.closeOnce.Do(func() {
		atomic.StoreInt32(&c.closeFlag, 1)
		close(c.heartbeatResetChan)
		close(c.closeChan)
		close(c.receiveChan)
		close(c.sendChan)
		_ = c.conn.Close()
		if h, ok := c.srv.eventHandlers[OnClose]; ok {
			go func() {
				defer func() {
					if err := recover(); err != nil {
						c.srv.log.Error(err)
					}
				}()
				err := h(newContext(c, nil))
				if err != nil {
					c.srv.log.Error(err)
					c.Close()
				}
			}()
		}
		c.onDisconnectCleanup()
		c.srv.log.Debugf("conn %s closed", c.RemoteAddr())
	})
}

func (c *connection) addWaitReplyChan(id uint64, waitReplyChan chan []byte) {
	c.waitReply.Store(id, waitReplyChan)
}

func (c *connection) removeWaitReplyChan(id uint64) {
	c.waitReply.Delete(id)
}

func (c *connection) IsClosed() bool {
	return atomic.LoadInt32(&c.closeFlag) == 1
}

func (c *connection) RemoteAddr() net.Addr {
	return c.conn.RemoteAddr()
}

func (c *connection) Get(key string) interface{} {
	v, _ := c.ctx.Load(key)
	return v
}

func (c *connection) Set(key string, value interface{}) {
	c.ctx.Store(key, value)
}

func (c *connection) checkHeartbeat() {
	defer func() {
		recover()
		c.srv.waitGroup.Done()
		c.Close()
	}()
	t := time.NewTicker(c.srv.opts.HeartbeatTimeout)
	for {
		select {
		case <-c.heartbeatResetChan:
			t.Reset(c.srv.opts.HeartbeatTimeout)
			continue
		case <-t.C:
			c.srv.log.Warn(fmt.Sprintf("%s timeout", c.RemoteAddr().String()))
			c.Close()
			return
		case <-c.closeChan:
			return
		}
	}
}

func (c *connection) readLoop() {
	defer func() {
		recover()
		c.srv.waitGroup.Done()
		c.Close()
	}()

	for {
		select {
		case <-c.srv.exitChan:
			return
		case <-c.closeChan:
			return
		default:
		}

		p, err := c.srv.opts.Protocol.ReadPacket(c.conn)
		if err != nil {
			if errors.Cause(err) == io.EOF {
				c.srv.log.Debug("client closed connection")
			} else {
				c.srv.log.Error(err)
			}
			if h, ok := c.srv.eventHandlers[OnDisconnect]; ok {
				go func() {
					defer func() {
						if err := recover(); err != nil {
							c.srv.log.Error(err)
						}
					}()
					err := h(newContext(c, nil))
					if err != nil {
						c.srv.log.Error(err)
						c.Close()
					}
				}()
			}
			return
		}

		c.receiveChan <- p
	}
}

func (c *connection) writeLoop() {
	defer func() {
		recover()
		c.srv.waitGroup.Done()
		c.Close()
	}()

	for {
		select {
		case <-c.srv.exitChan:
			return
		case <-c.closeChan:
			return
		case p := <-c.sendChan:
			if c.IsClosed() {
				return
			}
			if _, err := c.srv.opts.Protocol.WritePacket(p, c.conn); err != nil {
				c.srv.log.Error(err)
				return
			}
		}
	}
}

func (c *connection) handleLoop() {
	defer func() {
		recover()
		c.srv.waitGroup.Done()
		c.Close()
	}()

	for {
		select {
		case <-c.srv.exitChan:
			return
		case <-c.closeChan:
			return
		case p := <-c.receiveChan:
			if c.IsClosed() {
				return
			}
			c.heartbeatResetChan <- struct{}{}

			switch p.Key() {
			case protocol.MessageKeyHeartbeat:
				if h, ok := c.srv.eventHandlers[OnHeartbeat]; ok {
					go func() {
						defer func() {
							if err := recover(); err != nil {
								c.srv.log.Error(err)
							}
						}()
						err := h(newContext(c, nil))
						if err != nil {
							c.srv.log.Error(err)
							c.Close()
						}
					}()
				}
			case protocol.MessageKeyEmit, protocol.MessageKeyAckRequest:
				go func() {
					defer func() {
						if err := recover(); err != nil {
							c.srv.log.Error(err)
						}
					}()

					router := c.srv.findRouter(p.Event())
					h := router.Handler(p.Event())
					if h == nil {
						c.srv.log.Error("not find handler for event ", p.Event())
						return
					}
					h = applyMiddleware(h, router.middlewares()...)
					if router != c.srv {
						h = applyMiddleware(h, c.srv.middlewares()...)
					}

					err := h(newContext(c, p))
					if err != nil {
						c.srv.log.Error(err)
					}
				}()
			case protocol.MessageKeyAckResponse:
				go func() {
					if replyChan, ok := c.waitReply.Load(p.Id()); ok {
						replyChan.(chan []byte) <- p.Body()
					} else {
						c.srv.log.Error("not found reply channel for message: ", p.Id())
					}
				}()
			}

		}
	}
}

func (c *connection) reply(p protocol.Packet) (err error) {
	if c.IsClosed() {
		return ErrConnClosed
	}

	defer func() {
		if e := recover(); e != nil {
			err = ErrConnClosed
		}
	}()

	select {
	case c.sendChan <- p:
		return nil
	case <-c.closeChan:
		return ErrConnClosed
	case <-time.After(c.srv.opts.MessageTimeout):
		return ErrMessageTimeout
	}
}

func (c *connection) Emit(event string, v interface{}) (err error) {
	if c.IsClosed() {
		return ErrConnClosed
	}

	defer func() {
		if e := recover(); e != nil {
			err = ErrConnClosed
		}
	}()

	var p protocol.Packet

	if v == nil {
		p = c.srv.opts.Protocol.NewPacket(protocol.MessageKeyEmit, c.srv.idGenerator.Generate(), event, nil)
	} else {
		msg, err := c.srv.opts.Codec.Encode(v)
		if err != nil {
			return err
		}
		p = c.srv.opts.Protocol.NewPacket(protocol.MessageKeyEmit, c.srv.idGenerator.Generate(), event, msg)
	}

	select {
	case c.sendChan <- p:
		return nil
	case <-c.closeChan:
		return ErrConnClosed
	case <-time.After(c.srv.opts.MessageTimeout):
		return ErrMessageTimeout
	}
}

func (c *connection) Ack(event string, v interface{}, r interface{}) (err error) {
	if c.IsClosed() {
		return ErrConnClosed
	}

	defer func() {
		if e := recover(); e != nil {
			c.srv.log.Error(err)
		}
	}()

	var p protocol.Packet

	if v == nil {
		p = c.srv.opts.Protocol.NewPacket(protocol.MessageKeyAckRequest, c.srv.idGenerator.Generate(), event, nil)
	} else {
		msg, err := c.srv.opts.Codec.Encode(v)
		if err != nil {
			return err
		}
		p = c.srv.opts.Protocol.NewPacket(protocol.MessageKeyAckRequest, c.srv.idGenerator.Generate(), event, msg)
	}

	replyChan := make(chan []byte)
	c.addWaitReplyChan(p.Id(), replyChan)

	c.sendChan <- p

	select {
	case reply := <-replyChan:
		c.removeWaitReplyChan(p.Id())
		close(replyChan)
		return c.srv.opts.Codec.Decode(reply, r)
	case <-c.closeChan:
		return ErrConnClosed
	case <-time.After(c.srv.opts.MessageTimeout):
		c.removeWaitReplyChan(p.Id())
		close(replyChan)
		return ErrMessageTimeout
	}
}

func (c *connection) BroadcastToAll(event string, msg []byte) {
	c.srv.BroadcastToAll(event, msg)
}

func (c *connection) BroadcastTo(room, event string, msg []byte) {
	c.srv.BroadcastTo(room, event, msg)
}

func (c *connection) Join(room string) {

	c.srv.connsLock.Lock()
	defer c.srv.connsLock.Unlock()

	cn := c.srv.conns
	if _, ok := cn[room]; !ok {
		cn[room] = make(map[*connection]struct{})
	}

	byRoom := c.srv.rooms
	if _, ok := byRoom[c]; !ok {
		byRoom[c] = make(map[string]struct{})
	}

	cn[room][c] = struct{}{}
	byRoom[c][room] = struct{}{}
}

func (c *connection) Leave(room string) {

	c.srv.connsLock.Lock()
	defer c.srv.connsLock.Unlock()

	cn := c.srv.conns
	if _, ok := cn[room]; ok {
		delete(cn[room], c)
		if len(cn[room]) == 0 {
			delete(cn, room)
		}
	}

	byRoom := c.srv.rooms
	if _, ok := byRoom[c]; ok {
		delete(byRoom[c], room)
	}
}

func (c *connection) onDisconnectCleanup() {
	c.srv.connsLock.Lock()
	defer c.srv.connsLock.Unlock()

	cn := c.srv.conns
	byRoom, ok := c.srv.rooms[c]
	if ok {
		for room := range byRoom {
			if curRoom, ok := cn[room]; ok {
				delete(curRoom, c)
				if len(curRoom) == 0 {
					delete(cn, room)
				}
			}
		}

		delete(c.srv.rooms, c)
	}

	c.srv.sidsLock.Lock()
	defer c.srv.sidsLock.Unlock()

	delete(c.srv.sids, c.Id())
}

func generateSid(custom string) string {
	hash := fmt.Sprintf("%s %s %d %d", custom, time.Now(), rand.Uint32(), rand.Uint32())
	buf := bytes.NewBuffer(nil)
	sum := md5.Sum([]byte(hash))
	encoder := base64.NewEncoder(base64.URLEncoding, buf)
	_, _ = encoder.Write(sum[:])
	_ = encoder.Close()
	return buf.String()[:20]
}
